package net.xuele.learn.flink.keyedState;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

/**
 *
 * 有状态的转换
 *
 *
 * 为了实现这个功能，Deduplicator 需要记录每个键是否已经有了相应的记录。它将通过使用 Flink 的 keyed state 接口来做这件事
 *
 * 使用像这样的 keyed stream 的时候，Flink 会为每个状态中管理的条目维护一个键值存储
 *
 *
 */
public class Deduplicator extends  RichFlatMapFunction<Event, Event> {
    private static final long serialVersionUID = -4266511712407777574L;

    /**
     * 意思是对于 每个键 ，Flink 将存储一个单一的对象
     */
    ValueState<Boolean> keyHasBeenSeen;

    /**
     * 在算子初始化时调用一次，可以用来加载一些静态数据，或者建立外部服务的链接等
     * @param parameters
     * @throws Exception
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<Boolean> desc = new ValueStateDescriptor<>("keyHasBeenSeen", Types.BOOLEAN);
        // getRuntimeContext方法可以用于创建和访问flink管理的状态
        keyHasBeenSeen = getRuntimeContext().getState(desc);
    }



    @Override
    public void flatMap(Event event, Collector<Event> collector) throws Exception {
        // 第一次这个里面是没有值的
        // 对于当次来说，这里面的值是上一次进行设置的，并且在此之前是进行了keyBy操作的
        if (keyHasBeenSeen.value() == null) {
            collector.collect(event);
            keyHasBeenSeen.update(true);
        }
    }
}
